package com.app.kafka
import java.util.Date

import org.apache.commons.lang.time.FastDateFormat
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark._
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming._
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
import org.apache.spark.streaming.kafka010._

/**
 * kafka 分区偏移量  自动异步提交
 */
object KafkaItself {

  def main(args: Array[String]): Unit = {

    val conf = new SparkConf().setMaster("local[2]").setAppName("NetworkWordCount")
    conf.set("spark.streaming.kafka.maxRatePerPartition", "10000")

    val ssc = new StreamingContext(conf, Seconds(1))


    val kafkaParams = Map[String, Object](
      "bootstrap.servers" -> "localhost:9092,anotherhost:9092",
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> "hello",
      "auto.offset.reset" -> "latest",
      "enable.auto.commit" -> (false: java.lang.Boolean)
    )

    val topics = Array("topicA")
    val stream = KafkaUtils.createDirectStream[String, String](
      ssc,
      PreferConsistent,
      Subscribe[String, String](topics, kafkaParams)
    )

    val kstream: DStream[(String, Int)] = stream.transform(stream => {

      val rdd: RDD[(String, Int)] = stream.filter { record =>
        val message: String = record.value()
        null != message && message.trim.split(",").length == 4
      }.map { record =>
        val keyword: String = record.value().trim.split(",").last
        (keyword, 1)
      }.reduceByKey(_ + _)
      rdd
    })

    kstream.foreachRDD((rdd,time)=>{
      val batchTime: String = FastDateFormat.getInstance("yyyy/MM/dd HH:mm:ss")
        .format(new Date(time.milliseconds))
      println("-------------------------------------------")
      println(s"BatchTime: $batchTime")
      println("-------------------------------------------")
      if(!rdd.isEmpty()){
        rdd.coalesce(1).foreachPartition{_.foreach(println)}
      }

      //自动异步提交
      val offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
      // some time later, after outputs have completed
      stream.asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges)
    })

    ssc.start()
    ssc.awaitTermination()
    ssc.stop(true,true)
  }
}
